package com.intmall.flink.operator;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

public class SourceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 从文件中读取数据
        DataStreamSource<String> streamSource = env.readTextFile("input/clicks.txt");
//        streamSource.print();

        // 2. 从集合中读取数据
        List<Integer> nums = new ArrayList<>();
        nums.add(2);
        nums.add(3);
        nums.add(4);
        DataStreamSource<Integer> numStream = env.fromCollection(nums);
//        numStream.print("nums");

        ArrayList<Event> events = new ArrayList<>();
        events.add(new Event("Jim", "./home", 1000L));
        events.add(new Event("Join", "./cart", 2000L));
        events.add(new Event("Lili", "./order", 3000L));
        DataStreamSource<Event> eventStream = env.fromCollection(events);
//        eventStream.print("events");

        // 3. 从元素读取数据
        DataStreamSource<Event> elementStream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
//        elementStream.print("elements");

        // 4. 从Socket文本流读取(nc -lk 7777)
        // 吞吐量小、不稳定，适合测试
        DataStreamSource<String> socketStream = env.socketTextStream("hadoop101", 7777);
        socketStream.print("socketStream");

        env.execute();
    }
}
